package com.ehualu.javahadoop.until;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hdfs.DistributedFileSystem;

import java.io.IOException;
import java.net.URI;
import java.util.List;


/**
 * @Author: 吴敬超
 * @Date: 2020/3/2 13:05
 */
public class HdfsUntil {
    private final static Log LOG = LogFactory.getLog(HdfsUntil.class.getName());
    private static String PATH_TO_HDFS_SITE_XML = HdfsUntil.class.getClassLoader().getResource("hdfs-site.xml")
            .getPath();
    private static String PATH_TO_CORE_SITE_XML = HdfsUntil.class.getClassLoader().getResource("core-site.xml")
            .getPath();

    private static Configuration conf = null;

    private static void confLoad() throws IOException {

        conf = new Configuration();
        // conf file
        conf.addResource(new Path(PATH_TO_HDFS_SITE_XML));
        conf.addResource(new Path(PATH_TO_CORE_SITE_XML));
        System.setProperty("HADOOP_USER_NAME", "hdfs");
        // conf.addResource(new Path(PATH_TO_SMALL_SITE_XML));
    }

    /*
     *
     * 构造函数
     * 自动加载conf
     * */
    public HdfsUntil() throws IOException {
        confLoad();
//        instanceBuild();
    }

    /*
     *修复hdfs文件
     *
     * */
    public void recover(String firpath, List<String> scpaths) throws IOException {

        DistributedFileSystem fs = new DistributedFileSystem();

        for (String scpath : scpaths) {

            String finalpath = firpath + scpath;

            System.out.println("ffpat:" + finalpath);
            Path finalppath = new Path(finalpath);

            fs.initialize(URI.create(finalpath), conf);

            if (!fs.exists(finalppath)) {
                System.out.println("文件不存在跳出");

                continue;
            }

            FileStatus[] fileStatuses = fs.listStatus(new Path(finalpath));

            for (FileStatus fileStatus : fileStatuses) {
                String everypath = fileStatus.getPath().toString();
                System.out.println(everypath);
                Path evepath = new Path(everypath);
//                !fSystem.exists(filePath)

                if (fs.isFile(evepath)) {
                    System.out.println("是文件修复");
//                    LOG.info("是文件修复");
                    fs.recoverLease(evepath);

                } else {
                    System.out.println("是路径忽略");
//                    LOG.info("是路径忽略");

                }
            }

        }

        fs.close();
    }

}
